package com.tongcao.cn.mq.listner;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.*;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
import com.google.common.base.Strings;
import com.tongcao.cn.mq.to.AsyncMsgRM;
import com.xiaoleilu.hutool.lang.Assert;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetAddress;
import java.util.List;

/**
 * @Author:tongcao
 * @Date:2019/1/25 17:13
 * Description:
 */
public class ConsumerListenerForRm extends BaseConsumerListener {


    private Logger logger = LoggerFactory.getLogger(ConsumerListenerForRm.class);
    private DefaultMQPushConsumer consumer;                 // 消费者
    private String consumerGroup;                           // 改消费者所属的组
    private String namesrvAddr;                             // RocketMQ 的 NameSRV
    private MessageModel messageModel = MessageModel.CLUSTERING; // 消费模式。默认是集群模式
    //一个新的订阅组默认第一次启动从队列的最后位置开始消费
    private ConsumeFromWhere consumeFromWhere = ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET;
    public static volatile boolean suspend = false;//暂停消费，日常环境可用（可在控制台配置消费者的启停）
    //回溯消费时间
    private String consumeTimestamp;
    //是否启动顺序消费组
    private RegisterConsumeType registerConsumeType = RegisterConsumeType.CONCURRENTLY;
    //消费消息线程，最小数目
    private int consumeThreadMin = 20;

    /**
     * Listener启动
     */
    @Override
    void start() throws IOException, MQClientException {
        /**
         * 一个应用创建一个Consumer，由应用来维护此对象，可以设置为全局对象或者单例<br>
         * 注意：ConsumerGroupName需要由应用来保证唯一
         */
        Assert.isTrue(consumerGroup.startsWith("c_") && consumerGroup.contains(topic),
                "消费者不符合规范！consumerGroup:" + consumerGroup + ",topic:" + topic);
        //广播模式采用动态消费组的方式
        if (messageModel.equals(MessageModel.BROADCASTING)) {
            consumerGroup = consumerGroup + InetAddress.getLocalHost().getHostAddress().replace(".", "_");
        }
        //todo
        if (true && suspend) return;
        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeThreadMin(consumeThreadMin);
        //订阅指定topic下tags
        consumer.subscribe(topic, messageHandle.getSubExpression());
        /**
         * Consumer第一次启动默认从队列尾部开始消费
         * 如果非第一次启动，那么按照上次消费的位置继续消费
         */
        consumer.setConsumeFromWhere(consumeFromWhere);
        if (consumeFromWhere.equals(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP) &&
                !Strings.isNullOrEmpty(consumeTimestamp)) {
            consumer.setConsumeTimestamp(consumeTimestamp);
        }
        switch (registerConsumeType) {
            case ORDERLY:
                consumer.registerMessageListener(new BaseListenerOrderly());
                break;
            case CONCURRENTLY:
            default:
                consumer.registerMessageListener(new BaseListenerConcurrently());
                break;
        }
        consumer.start();
        logger.info("ConsumerListenerForRm started!topic:" + topic + ",expression:" + messageHandle.getSubExpression() + "  consumerGroup:" + consumerGroup + "   namesrvAddr:" + namesrvAddr);
    }


    public void setLogger(Logger logger) {
        this.logger = logger;
    }

    public void setConsumer(DefaultMQPushConsumer consumer) {
        this.consumer = consumer;
    }

    public void setConsumerGroup(String consumerGroup) {
        this.consumerGroup = consumerGroup;
    }

    public void setNamesrvAddr(String namesrvAddr) {
        this.namesrvAddr = namesrvAddr;
    }

    public void setMessageModel(MessageModel messageModel) {
        this.messageModel = messageModel;
    }

    public void setConsumeFromWhere(ConsumeFromWhere consumeFromWhere) {
        this.consumeFromWhere = consumeFromWhere;
    }

    public static void setSuspend(boolean suspend) {
        ConsumerListenerForRm.suspend = suspend;
    }

    public void setConsumeTimestamp(String consumeTimestamp) {
        this.consumeTimestamp = consumeTimestamp;
    }

    public void setRegisterConsumeType(RegisterConsumeType registerConsumeType) {
        this.registerConsumeType = registerConsumeType;
    }

    public void setConsumeThreadMin(int consumeThreadMin) {
        this.consumeThreadMin = consumeThreadMin;
    }

    @Override
    void close() {
        this.consumer.shutdown();
        logger.info("ConsumerListenerForRm closed !");
    }

    public enum RegisterConsumeType {
        ORDERLY, CONCURRENTLY,
    }

    private class BaseListenerOrderly implements MessageListenerOrderly {
        @Override
        public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
            MessageExt msg = msgs.get(0);
            AsyncMsgRM asyncMsgRM = new AsyncMsgRM(msg);
            boolean ret = messageHandle.consume(asyncMsgRM);
            if (!ret) {
                logger.error("consume MQ failed,msgID:" + msg.getMsgId());
            }
            return ret ? ConsumeOrderlyStatus.SUCCESS :
                    ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
        }
    }


    private class BaseListenerConcurrently implements MessageListenerConcurrently {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                        ConsumeConcurrentlyContext context) {
            MessageExt msg = msgs.get(0);
            AsyncMsgRM asyncMsgRM = new AsyncMsgRM(msg);
            boolean ret = messageHandle.consume(asyncMsgRM);
            if (!ret) {
                logger.error("consume MQ failed,msgID:" + msg.getMsgId());
            } else {
                logger.debug("consume MQ,msg:" + msg.toString() + ",handle result:success.");
            }
            return ret ? ConsumeConcurrentlyStatus.CONSUME_SUCCESS :
                    ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }

}
